package com.wwjd.starter.canal.client.abstracts;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.wwjd.starter.canal.client.interfaces.CanalClient;
import com.wwjd.starter.canal.client.interfaces.TransponderFactory;
import com.wwjd.starter.canal.client.transfer.DefaultMessageTransponder;
import com.wwjd.starter.canal.config.CanalProperties;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.StringUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;

/**
 * Canal 客户端抽象类
 *
 * @author 阿导
 * @CopyRight 萬物皆導
 * @created 2018/5/28 15:02
 * @Modified_By 阿导 2018/5/28 15:02
 */
public abstract class AbstractCanalClient implements CanalClient {
    /**
     * 运行状态
     */
    private volatile boolean running;

    /**
     * canal 配置
     */
    private CanalProperties canalProperties;

    /**
     * 转换工厂类
     */
    protected final TransponderFactory factory;

    /**
     * 构造方法，初始化 canal 的配置以及转换信息的工厂实例
     *
     * @param canalProperties
     * @return
     * @author 阿导
     * @time 2018/5/28 15:04
     * @CopyRight 万物皆导
     */
    protected AbstractCanalClient(CanalProperties canalProperties) {
        Objects.requireNonNull(canalProperties, "canalConfig 不能为空");
        this.canalProperties = canalProperties;
        this.factory = (connector, config, listeners, annoListeners) -> new DefaultMessageTransponder(connector, config, listeners, annoListeners);
    }

    /**
     * 别拦我，我想开启 canal 客户端
     *
     * @param
     * @return
     * @author 阿导
     * @time 2018/5/28 15:05
     * @CopyRight 万物皆导
     */
    @Override
    public void start() {
        if (canalProperties == null || MapUtils.isEmpty(canalProperties.getInstances())) {
            throw new CanalClientException("无法解析 canal 的连接信息，请联系开发人员!");
        }
        // 可能有多个客户端
        Map<String, CanalProperties.Instance> instanceMap = canalProperties.getInstances();
        for (Map.Entry<String, CanalProperties.Instance> instanceEntry : instanceMap.entrySet()) {
            CanalProperties.Instance instance = instanceEntry.getValue();
            CanalConnector connector;
            if (instance.isClusterEnabled()) {
                List<SocketAddress> addresses = new ArrayList<>();
                for (String zookeeperAddress : instance.getZookeeperAddress()) {
                    String[] entry = zookeeperAddress.split(":");
                    if (entry.length != 2) {
                        throw new CanalClientException("zookeeper 地址 " + zookeeperAddress + "格式不正确，应该为 ip1:port1,ip2:port2 形式");
                    }
                    // 若符合设定规则，先加入集合
                    InetSocketAddress inetSocketAddress = new InetSocketAddress(entry[0], Integer.parseInt(entry[1]));
                    addresses.add(inetSocketAddress);
                }
                // 若集群的话，使用 newClusterConnector 方法初始化
                connector = CanalConnectors.newClusterConnector(addresses, instanceEntry.getKey(), instance.getUserName(), instance.getPassword());
            } else {
                // 若不是集群的话，使用 newSingleConnector 初始化
                InetSocketAddress inetSocketAddress = new InetSocketAddress(instance.getHost(), instance.getPort());
                connector = CanalConnectors.newSingleConnector(inetSocketAddress, instanceEntry.getKey(), instance.getUserName(), instance.getPassword());
            }
            connector.connect();
            if (StringUtils.isNotBlank(instance.getFilter())) {
                // canal 连接订阅，包含过滤规则
                connector.subscribe(instance.getFilter());
            } else {
                // canal 连接订阅，无过滤规则
                connector.subscribe();
            }
            connector.rollback();
            this.process(connector, instanceEntry);
        }
    }

    /**
     * 初始化 canal 连接
     *
     * @param connector
     * @param config
     * @return
     * @author 阿导
     * @time 2018/5/28 15:06
     * @CopyRight 万物皆导
     */
    protected abstract void process(CanalConnector connector, Map.Entry<String, CanalProperties.Instance> config);

    /**
     * 停止 canal 客户端
     *
     * @param
     * @return
     * @author 阿导
     * @time 2018/5/28 15:08
     * @CopyRight 万物皆导
     */
    @Override
    public void stop() {
        this.running = false;
    }

    /**
     * 返回 canal 客户端的状态
     *
     * @param
     * @return
     * @author 阿导
     * @time 2018/5/28 15:08
     * @CopyRight 万物皆导
     */
    @Override
    public boolean isRunning() {
        return running;
    }
}